package com.kk.utils;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.function.Supplier;

/**
 * 虚拟线程工具类 - 基于 Java 21 Virtual Thread
 * 提供虚拟线程的创建、执行、批量处理等功能
 *
 * @author Your Name
 * @since Java 21
 */
@Component
public class VirtualThreadUtil {

    private static final Logger log = LoggerFactory.getLogger(VirtualThreadUtil.class);

    /**
     * 创建虚拟线程执行器（每个任务一个新虚拟线程）
     */
    private static final ExecutorService VIRTUAL_THREAD_EXECUTOR =
            Executors.newVirtualThreadPerTaskExecutor();

    /**
     * 创建结构化任务作用域的执行器（用于有限的并发控制）
     */
    public static ExecutorService createVirtualThreadExecutor() {
        return Executors.newVirtualThreadPerTaskExecutor();
    }

    /**
     * 创建指定并发数的虚拟线程池（使用信号量控制并发）
     */
    public static ExecutorService createBoundedVirtualThreadExecutor(int maxConcurrency) {
        return new ThreadPoolExecutor(
                0,
                Integer.MAX_VALUE,
                0L,
                TimeUnit.MILLISECONDS,
                new SynchronousQueue<>(),
                Thread.ofVirtual().factory()
        );
    }

    /**
     * 异步执行单个任务（无返回值）
     */
    public static CompletableFuture<Void> runAsync(Runnable task) {
        return CompletableFuture.runAsync(task, VIRTUAL_THREAD_EXECUTOR);
    }

    /**
     * 异步执行单个任务（有返回值）
     */
    public static <T> CompletableFuture<T> supplyAsync(Supplier<T> supplier) {
        return CompletableFuture.supplyAsync(supplier, VIRTUAL_THREAD_EXECUTOR);
    }

    /**
     * 批量执行任务（无返回值）
     */
    public static void executeAll(List<Runnable> tasks) {
        List<CompletableFuture<Void>> futures = tasks.stream()
                .map(task -> CompletableFuture.runAsync(task, VIRTUAL_THREAD_EXECUTOR))
                .toList();

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }

    /**
     * 批量执行任务（有返回值）
     */
    public static <T> List<T> executeAllWithResult(List<Callable<T>> tasks) {
        try {
            return VIRTUAL_THREAD_EXECUTOR.invokeAll(tasks).stream()
                    .map(future -> {
                        try {
                            return future.get();
                        } catch (InterruptedException | ExecutionException e) {
                            log.error("Task execution failed", e);
                            Thread.currentThread().interrupt();
                            return null;
                        }
                    })
                    .toList();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Batch execution interrupted", e);
            return List.of();
        }
    }

    /**
     * 批量执行任务（有返回值，带超时）
     */
    public static <T> List<T> executeAllWithResult(List<Callable<T>> tasks, Duration timeout) {
        try {
            return VIRTUAL_THREAD_EXECUTOR.invokeAll(tasks, timeout.toMillis(), TimeUnit.MILLISECONDS)
                    .stream()
                    .map(future -> {
                        try {
                            return future.get();
                        } catch (InterruptedException | ExecutionException | CancellationException e) {
                            log.error("Task execution failed or timeout", e);
                            Thread.currentThread().interrupt();
                            return null;
                        }
                    })
                    .toList();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Batch execution interrupted", e);
            return List.of();
        }
    }

    /**
     * 使用结构化并发（Structured Concurrency）执行任务
     * Java 21 预览特性，需要添加 --enable-preview
     */
    public static <T> List<T> executeWithStructuredConcurrency(List<Callable<T>> tasks) {
        List<T> results = new ArrayList<>();

        // 注意：这里使用传统方式，如果需要使用 StructuredTaskScope，需要启用预览特性
        List<CompletableFuture<T>> futures = tasks.stream()
                .map(task -> CompletableFuture.supplyAsync(() -> {
                    try {
                        return task.call();
                    } catch (Exception e) {
                        log.error("Task execution failed", e);
                        return null;
                    }
                }, VIRTUAL_THREAD_EXECUTOR))
                .toList();

        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

        futures.forEach(future -> {
            try {
                T result = future.get();
                if (result != null) {
                    results.add(result);
                }
            } catch (InterruptedException | ExecutionException e) {
                log.error("Failed to get result", e);
                Thread.currentThread().interrupt();
            }
        });

        return results;
    }

    /**
     * 并行处理集合（使用虚拟线程）
     */
    public static <T, R> List<R> parallelProcess(List<T> items,
                                                 java.util.function.Function<T, R> processor) {
        List<Callable<R>> tasks = items.stream()
                .map(item -> (Callable<R>) () -> processor.apply(item))
                .toList();

        return executeAllWithResult(tasks);
    }

    /**
     * 并行处理集合（带超时）
     */
    public static <T, R> List<R> parallelProcess(List<T> items,
                                                 java.util.function.Function<T, R> processor,
                                                 Duration timeout) {
        List<Callable<R>> tasks = items.stream()
                .map(item -> (Callable<R>) () -> processor.apply(item))
                .toList();

        return executeAllWithResult(tasks, timeout);
    }

    /**
     * 创建并启动虚拟线程
     */
    public static Thread startVirtualThread(Runnable task) {
        return Thread.startVirtualThread(task);
    }

    /**
     * 创建虚拟线程（未启动）
     */
    public static Thread buildVirtualThread(String name, Runnable task) {
        return Thread.ofVirtual()
                .name(name)
                .unstarted(task);
    }

    /**
     * 检查当前线程是否为虚拟线程
     */
    public static boolean isVirtualThread() {
        return Thread.currentThread().isVirtual();
    }

    /**
     * 获取当前线程信息
     */
    public static String getCurrentThreadInfo() {
        Thread thread = Thread.currentThread();
        return String.format("Thread[name=%s, id=%d, virtual=%s]",
                thread.getName(),
                thread.threadId(),
                thread.isVirtual());
    }

    /**
     * 优雅关闭执行器
     */
    public static void shutdown() {
        VIRTUAL_THREAD_EXECUTOR.shutdown();
        try {
            if (!VIRTUAL_THREAD_EXECUTOR.awaitTermination(60, TimeUnit.SECONDS)) {
                VIRTUAL_THREAD_EXECUTOR.shutdownNow();
            }
        } catch (InterruptedException e) {
            VIRTUAL_THREAD_EXECUTOR.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 立即关闭执行器
     */
    public static void shutdownNow() {
        VIRTUAL_THREAD_EXECUTOR.shutdownNow();
    }
}